package web

import (
	"bufio"
	"gitee.com/yysf_xin/dplog/pkg/config"
	"gitee.com/yysf_xin/dplog/pkg/slog"
	"gitee.com/yysf_xin/dplog/pkg/web/pb"
	"gitee.com/yysf_xin/go-common/tools"
	"github.com/fsnotify/fsnotify"
	"google.golang.org/protobuf/encoding/protowire"
	"google.golang.org/protobuf/proto"
	"io"
	"os"
	"path/filepath"
	"time"
)

type Watcher interface {
}
type Event interface {
	GetType() int
}

type FileWatch struct {
	Id      int
	clients map[*client]bool
	p       *config.Pattern
	c       *slog.Collect

	rd        CountReader
	w         *DirWatch
	flushTime time.Time
	curr      string

	seq int32
	msg pb.Message
	res pb.Response
}

type DirWatch struct {
	Id      int
	Dir     *config.Dir
	w       *fsnotify.Watcher
	l       int
	c       *config.ServerConfig
	files   []*FileWatch
	e       chan Event
	marshal proto.MarshalOptions
}

func (d *FileWatch) FireMsg(ref *slog.MessageRef) {
	d.seq++
	//tools.Log.Println("fireMsg", ref.Meta.Offset)
	if len(d.clients) == 0 {
		return
	}
	if d.p.Type == "java" {
		fx := ref.GetRef().(*slog.MessageImpl)
		d.msg.Time = fx.GetTime().UnixMilli()
		d.msg.AppName = fx.AppName
		d.msg.Trace = fx.TraceID
		d.msg.Logger = fx.Logger
		d.msg.Method = fx.Method
		d.msg.Message = fx.Message
		d.msg.SubMessage = fx.SubMessage
	} else {
		d.msg.Message = tools.BytesToString(ref.Buf.Bytes())
		d.msg.Level = 1
		d.msg.Time = time.Now().UnixMilli()
	}
	d.msg.Id = d.seq

	var (
		bufP = bufPool.Get().([]byte)
		bufM = bufPool.Get().([]byte)
		err  error
	)
	bufP = bufP[:0]
	bufM = bufM[:0]
	bufM, err = d.w.marshal.MarshalAppend(bufM, &d.msg)

	if err == nil {
		d.res.Code = 200
		d.res.Type = pb.MsgType_LOG_MSG
		d.res.Data = bufM
		bufP = protowire.AppendVarint(bufP, uint64(d.w.marshal.Size(&d.res)))
		bufP, err = d.w.marshal.MarshalAppend(bufP, &d.res)
		if err == nil {
			for k := range d.clients {
				_, err = k.Write(bufP)
			}
		}
	}
	bufPool.Put(bufM)
	bufPool.Put(bufP)
}

type CountReader struct {
	f      *os.File
	r      *bufio.Reader
	offset int64
	m      slog.MetaData
}

func (c *CountReader) Read(p []byte) (n int, err error) {
	n, err = c.f.Read(p)
	c.offset += int64(n)
	return
}

func (c *CountReader) Reset(f *os.File) {
	if c.f == f {
		return
	}
	if c.f != nil {
		c.offset = 0
		_ = c.f.Close()
	} else {
		off, err := f.Seek(0, io.SeekEnd)
		if err != nil {
			tools.Log.Println(err)
		}
		c.offset = off
	}

	c.f = f
	if c.r == nil {
		c.r = bufio.NewReader(c)
	} else {
		c.r.Reset(c)
	}

	return
}

func (c *CountReader) GetOffset() int64 {
	return c.offset - int64(c.r.Buffered())
}

func (c *CountReader) Close() (err error) {
	if c.f != nil {
		err = c.f.Close()
		c.f = nil
	}
	return
}

func (d *DirWatch) Close() (err error) {
	err = d.w.Close()
	for _, v := range d.files {
		err = v.rd.Close()
		if err != nil {
			tools.Log.Println(err)
		}
	}
	close(d.e)
	return
}

func NewFileWatch(pattern *config.Pattern, c *slog.Collect) (f *FileWatch, err error) {
	f = &FileWatch{
		p:       pattern,
		clients: make(map[*client]bool),
		c:       c,
	}
	c.Emitter = f
	return
}

func NewDirWatch(dir *config.Dir, c *config.ServerConfig) (d *DirWatch, err error) {
	dir.Dir = filepath.Clean(dir.Dir)
	w, err := fsnotify.NewWatcher()
	if err == nil {
		err = w.Add(dir.Dir)
	}
	if err != nil {
		tools.Log.Println("watch dir err", dir.Dir)
		return
	}
	d = &DirWatch{
		Dir:   dir,
		w:     w,
		l:     len(dir.Dir),
		c:     c,
		files: make([]*FileWatch, len(dir.Items)),
		e:     make(chan Event, 10),
	}

	return
}

type watchCtx struct {
	client  *client
	id      int
	unWatch bool
}

func (w *watchCtx) GetType() int {
	return evtAttach
}

func (d *DirWatch) Run() {
	ti := time.NewTicker(1 * time.Second)
	defer ti.Stop()
	for {
		select {
		case _ = <-ti.C:
			for _, v := range d.files {
				v.c.Flush()
				for c := range v.clients {
					c.Flush()
				}
			}
		case e, ok := <-d.w.Events:
			if !ok {
				return
			}
			name := e.Name[d.l+1:]
			//tools.Log.Println("write event", name)
			for _, v := range d.files {
				if r, _ := filepath.Match(v.p.Pattern, name); r {
					v.notify(name, e)
					break
				}
			}
		case e, ok := <-d.w.Errors:
			if !ok {
				return
			}
			tools.Log.Println("error", e)

		case e, ok := <-d.e:
			if !ok {
				return
			}
			switch e.GetType() {
			case evtAttach:
				ev := e.(*watchCtx)
				if ev.unWatch {
					tools.Log.Println("client unwatch", d.files[ev.id].p.Pattern)
					delete(d.files[ev.id].clients, ev.client)
				} else {
					tools.Log.Println("client watch", d.files[ev.id].p.Pattern)

					d.files[ev.id].clients[ev.client] = true
				}
			}
		}

	}
}

func (d *FileWatch) reInitReader(name string) (err error) {
	if name == "" {
		name = d.rd.f.Name()
	} else {
		name = filepath.Join(d.w.Dir.Dir, name)
	}
	tools.Log.Println("reInit file", name)
	f, err := os.Open(name)
	if err == nil {
		d.rd.Reset(f)
		d.c.Init(&d.rd.m)
		d.seq = 0
	}
	return
}

func (d *FileWatch) doRead() {
	count := 0
	d.rd.m.File = d.curr
	d.rd.m.Env = config.EnvStr
	for {
		d.rd.m.Offset = d.rd.GetOffset()
		line, prefix, err := d.rd.r.ReadLine()
		if err != nil {
			if err == io.EOF {
				err = nil
				//if count == 0 {
				//	err = d.reInitReader("")
				//}
			}
			if err != nil {
				tools.Log.Println(err)
				err = d.rd.Close()
			}
			break
		}
		count++
		d.rd.m.Line = count
		d.c.OnRead(line, prefix)
	}

}

func (d *FileWatch) notify(name string, e fsnotify.Event) {
	if e.Op&fsnotify.Write != fsnotify.Write {
		tools.Log.Println("event", e.String())
	}
	if d.curr == name {
		if e.Op&fsnotify.Create == fsnotify.Create {
			if d.rd.f != nil {
				d.doRead()
			}
			// re init on next
			d.curr = ""
			return
		}
		if e.Op&fsnotify.Remove == fsnotify.Remove {
			d.curr = ""
			return
		}
		if e.Op&fsnotify.Write == fsnotify.Write {
			if d.rd.f != nil {
				d.doRead()
				// d.flushTime = time.Now().Add(1 * time.Second)
			}
		}
	} else {
		if e.Op&fsnotify.Write == fsnotify.Write {
			d.curr = name
			err := d.reInitReader(name)
			if err != nil {
				tools.Log.Println(err)
			}
		}
	}

}
